package main

import (
	"bytes"
	"compress/gzip"
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/IBM/sarama"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/service/s3"
)

const (
	kafkaBrokers        = "localhost:9092"
	topic               = "device-info"
	s3Bucket            = "aws-aspen-s3-bucket"
	filePrefix          = "logs/"
	datetimeLayout      = "2006-01-02-15" // YYYYMMDD-HH
	uploadCheckInterval = 1 * time.Minute // 检查间隔
	uploadDelayMinutes  = 55              // 第10分钟上传
)

var (
	awsCfg       aws.Config
	s3Client     *s3.Client
	currentFiles = make(map[string]*os.File) // 跟踪当前打开的文件
	gzipWriters  = make(map[string]*gzip.Writer)
	//fileMsgNum     = make(map[string]int)
	lastUploadHour int // 记录上次上传的小时数
)

// 修改Message结构体为DeviceInfo
type DeviceInfo struct {
	Publisher      string `json:"publisher"`
	Bundle         string `json:"bundle"`
	Geo            string `json:"geo"`
	DeviceType     int    `json:"device_type"` // 根据实际类型调整
	Make           string `json:"make"`
	Model          string `json:"model"`
	OS             string `json:"os"`
	OSV            string `json:"osv"`
	Language       string `json:"language"`
	ConnectionType int    `json:"connectionType"` // 根据实际类型调整
	DeviceID       string `json:"device_id"`
	IP             string `json:"ip"`
	IPv6           string `json:"ipv6"`
	UserAgent      string `json:"user_agent"`
	Timestamp      int64  `json:"timestamp"`
}

func main() {
	// 初始化AWS配置
	var err error
	awsCfg, err = config.LoadDefaultConfig(context.TODO())
	if err != nil {
		log.Fatalf("AWS config error: %v", err)
	}
	s3Client = s3.NewFromConfig(awsCfg)

	// 配置Kafka消费者
	config := sarama.NewConfig()
	config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{
		sarama.NewBalanceStrategyRange(), // 保持原有Range分区分配策略
	}
	config.Consumer.Offsets.Initial = sarama.OffsetNewest

	consumer, err := sarama.NewConsumerGroup([]string{kafkaBrokers}, "s3-uploader", config)
	if err != nil {
		log.Fatalf("Error creating consumer group: %v", err)
	}
	defer consumer.Close()

	go scheduleHourlyUpload()
	// 启动消费循环
	handler := &consumerHandler{}
	for {
		err := consumer.Consume(context.Background(), []string{topic}, handler)
		if err != nil {
			log.Printf("Consume error: %v", err)
			time.Sleep(5 * time.Second)
		}
	}
}

type consumerHandler struct{}

func (h *consumerHandler) Setup(sarama.ConsumerGroupSession) error   { return nil }
func (h *consumerHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }
func (h *consumerHandler) ConsumeClaim(session sarama.ConsumerGroupSession,
	claim sarama.ConsumerGroupClaim) error {

	for msg := range claim.Messages() {
		var deviceInfo DeviceInfo
		if err := json.Unmarshal(msg.Value, &deviceInfo); err != nil {
			log.Printf("Error decoding message: %v", err)
			continue
		}

		// 处理消息
		if err := processMessage(deviceInfo); err != nil {
			log.Printf("Error processing message: %v", err)
		}
		session.MarkMessage(msg, "")
	}
	return nil
}

func processMessage(msg DeviceInfo) error {
	// 根据时间戳生成文件名
	t := time.UnixMilli(msg.Timestamp)
	timeKey := t.Format(datetimeLayout)
	fileKey := fmt.Sprintf("%s_%s_%s", msg.Geo, msg.OS, timeKey)

	// 获取或创建文件写入器
	writer, exists := gzipWriters[fileKey]
	if !exists {
		if err := createNewFile(fileKey, t); err != nil {
			return err
		}
		writer = gzipWriters[fileKey]

	}

	// 写入消息
	var builder strings.Builder
	builder.WriteString(msg.Publisher)
	builder.WriteString("@")
	builder.WriteString(msg.Bundle)
	builder.WriteString("@")
	builder.WriteString(msg.Geo)
	builder.WriteString("@")
	builder.WriteString(strconv.Itoa(msg.DeviceType))
	builder.WriteString("@")
	builder.WriteString(msg.Make)
	builder.WriteString("@")
	builder.WriteString(msg.Model)
	builder.WriteString("@")
	builder.WriteString(msg.OS)
	builder.WriteString("@")
	builder.WriteString(msg.OSV)
	builder.WriteString("@")
	builder.WriteString(msg.Language)
	builder.WriteString("@")
	builder.WriteString(strconv.Itoa(msg.ConnectionType))
	builder.WriteString("@")
	builder.WriteString(msg.DeviceID)
	builder.WriteString("@")
	builder.WriteString(msg.IP)
	builder.WriteString("@")
	builder.WriteString(msg.IPv6)
	builder.WriteString("@")
	builder.WriteString(msg.UserAgent)
	builder.WriteString("@")
	builder.WriteString(strconv.FormatInt(msg.Timestamp, 10))

	data := []byte(builder.String())
	//data, err := json.Marshal(msg)
	//if err != nil {
	//	return fmt.Errorf("marshal error: %w", err)
	//}

	// 写入消息（添加换行符）
	if _, err := writer.Write(append(data, '\n')); err != nil {
		return err
	}

	//i := fileMsgNum[fileKey]
	//fileMsgNum[fileKey] = i + 1
	//log.Printf("写入消息: %s, %d", fileKey, fileMsgNum[fileKey])

	// 检查是否需要滚动文件（每小时）
	//if time.Since(t) > time.Hour {
	//	if err := rotateFile(fileKey); err != nil {
	//		return err
	//	}
	//}
	return nil
}

func createNewFile(fileKey string, t time.Time) error {
	// 创建本地文件
	filename := fmt.Sprintf("%s.log.gz", fileKey)
	//filePath := filepath.Join(os.TempDir(), filename)
	filePath := filepath.Join("/home/ec2-user/data", filename)

	log.Printf("创建文件: %s, %s", filePath, filePath)

	file, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		return err
	}

	// 创建gzip writer
	gz := gzip.NewWriter(file)

	// 更新跟踪记录
	currentFiles[fileKey] = file
	gzipWriters[fileKey] = gz
	return nil
}

func uploadPreviousHourFiles(targetTime time.Time) {
	// 生成目标小时的文件名前缀
	timeKey := targetTime.Format(datetimeLayout)
	pattern := fmt.Sprintf("*_*_%s.log.gz", timeKey)

	// 查找匹配的本地文件
	//files, err := filepath.Glob(filepath.Join(os.TempDir(), pattern))
	files, err := filepath.Glob(filepath.Join("/home/ec2-user/data", pattern))
	if err != nil {
		log.Printf("文件查找失败: %v", err)
		return
	}

	// 并行上传文件
	for _, fpath := range files {
		go func(path string) {
			log.Printf("上传文件: %s", path)
			// 从路径中提取fileKey
			filename := filepath.Base(path)
			fileKey := filename[:len(filename)-len(".log.gz")]

			// 关闭并上传（如果文件仍处于打开状态）
			if writer, exists := gzipWriters[fileKey]; exists {
				writer.Close()
				delete(gzipWriters, fileKey)
			}
			if file, exists := currentFiles[fileKey]; exists {
				file.Close()
				delete(currentFiles, fileKey)
			}

			// 执行上传
			if err := uploadToS3(path, fileKey); err != nil {
				log.Printf("上传失败 %s: %v", path, err)
			} else {
				os.Remove(path)
			}
		}(fpath)
	}
}

// 新增定时调度函数
func scheduleHourlyUpload() {
	ticker := time.NewTicker(uploadCheckInterval)
	defer ticker.Stop()

	for range ticker.C {
		now := time.Now()
		// 每小时的10分钟触发（例如01:10, 02:10等）

		log.Printf("当前时间: %d", now.Minute())
		if now.Minute() == uploadDelayMinutes && now.Hour() != lastUploadHour {
			//targetHour := now.Add(-time.Hour) // 处理上个小时的数据
			uploadPreviousHourFiles(now)
			lastUploadHour = now.Hour()
		}
	}
}

// 独立的上传函数
func uploadToS3(filePath, fileKey string) error {
	data, err := os.ReadFile(filePath)
	if err != nil {
		return err
	}

	// 从fileKey中提取出geo, os, time
	parts := strings.Split(fileKey, "_")
	geo := parts[0]
	os := parts[1]
	timeKey := parts[2]
	// 从timeKey中提取出到日的部分  比如2025-03-10
	timeToDay := timeKey[:10]
	// TODO: 暂时模仿online2的路径  只用us region
	uploadFilePrefix := filePrefix + timeToDay + "/" + strings.ToLower(geo) + "/" + os + "/" + timeKey + "." + os + "." + strings.ToLower(geo) + ".us/" + fileKey
	_, err = s3Client.PutObject(context.TODO(), &s3.PutObjectInput{
		Bucket:      aws.String(s3Bucket),
		Key:         aws.String(uploadFilePrefix + ".log.gz"),
		Body:        bytes.NewReader(data),
		ContentType: aws.String("application/gzip"),
	})
	return err
}

//func rotateFile(fileKey string) error {
//	// 关闭当前文件
//	if err := gzipWriters[fileKey].Close(); err != nil {
//		return err
//	}
//	if err := currentFiles[fileKey].Close(); err != nil {
//		return err
//	}
//
//	// 上传到S3
//	filePath := filepath.Join(os.TempDir(), fileKey+".log.gz")
//	data, err := os.ReadFile(filePath)
//	if err != nil {
//		return err
//	}
//
//	// S3上传参数
//	_, err = s3Client.PutObject(context.TODO(), &s3.PutObjectInput{
//		Bucket:      aws.String(s3Bucket),
//		Key:         aws.String(filePrefix + fileKey + ".log.gz"),
//		Body:        bytes.NewReader(data),
//		ContentType: aws.String("application/gzip"),
//	})
//
//	// 清理本地文件
//	delete(currentFiles, fileKey)
//	delete(gzipWriters, fileKey)
//	os.Remove(filePath)
//
//	return err
//}
